import contextlib
import hashlib
import multiprocessing
import os
import queue
import random
import re
import resource
import shutil
import signal
import subprocess
import sys
import tempfile
import threading
import time
import traceback
from datetime import datetime, timedelta
from pathlib import Path
from multiprocessing import Event, Manager
import collections
import cv2
import fitz  # PyMuPDF
import numpy as np
import openpyxl
import psutil
import xlrd
from docx import Document
from elasticsearch import Elasticsearch
from openpyxl import Workbook
from paddleocr import PaddleOCR
from PIL import Image
from xlrd import xldate_as_datetime

# 配置常量
CPU_USEAGE = 0.7
MAX_RETRIES = 2
MAX_ES_RETRIES = 3
ES_RETRY_DELAY = 5
FILE_HANDLE_LIMIT = 65536
LOCK_TIMEOUT = 1200
TASK_BATCH_SIZE = 100  # 每次从ES获取的任务批大小
TASK_QUEUE_SIZE = 100  # 任务队列最大大小
RESULT_QUEUE_SIZE = 100  # 结果队列最大大小
TASKS_PER_WORKER = 1  # 每个工作进程处理的任务数

# Elasticsearch 配置
ES_HOST = "https://localhost:9200"
ES_API_KEY = "T0lqM2RKWUI4clBKeFVBenFaZGc6MTBWZmU3d3VSX0dHQy1iNjBfbHRyUQ=="
ES_CA_CERTS = "/srv/elasticsearch-8.12.2/config/certs/http_ca.crt"

# 支持的文件扩展名
SUPPORTED_EXTS = {
    '.xlsx', '.xls', '.docx', '.doc', 
    '.pptx', '.ppt', '.pdf', 
    '.html', '.htm', '.txt'
}

class ProcessPhase:
    def __init__(self):
        self.shutdown_flag = False
        self.force_shutdown = False
        self.setup_signal_handlers()
        self.es_client = self.init_elasticsearch()
        self.file_hash_index_name = "file_hash"
        self.check_xlrd_version()
        self.shutdown_event = multiprocessing.Event()
        self._next_worker_id = 0
        
        # 线程属性初始化
        self.progress_thread = None
        self.dispatcher_thread = None
        self.result_handler_thread = None
        
        # 使用Manager创建共享状态
        self.manager = Manager()
        
        # 共享计数器
        self.total_tasks = multiprocessing.Value('i', 0)          # 总任务数
        self.completed_tasks = multiprocessing.Value('i', 0)     # 已完成任务数
        self.success_tasks = multiprocessing.Value('i', 0)       # 成功任务数
        self.failed_tasks = multiprocessing.Value('i', 0)         # 失败任务数
        self.task_counter = multiprocessing.Value('i', 0)         # 已分发任务计数器
        
        # 事件队列
        self.event_queue = multiprocessing.Queue(maxsize=50)
        
        # 任务队列和结果队列
        self.task_queue = multiprocessing.Queue(maxsize=TASK_QUEUE_SIZE)
        self.result_queue = multiprocessing.Queue(maxsize=RESULT_QUEUE_SIZE)
        
        # 设置文件描述符限制
        try:
            resource.setrlimit(
                resource.RLIMIT_NOFILE, 
                (FILE_HANDLE_LIMIT, FILE_HANDLE_LIMIT)
            )
        except (ValueError, resource.error) as e:
            print(f"警告: 设置文件描述符限制失败: {str(e)}")
            
        self.memory_monitor_thread = threading.Thread(
            target=self._memory_monitor,
            daemon=True
        )
        self.memory_monitor_thread.start()
        self.event_queue.put("内存监控线程已启动")
        
    def check_xlrd_version(self):
        if hasattr(xlrd, '__VERSION__') and xlrd.__VERSION__ >= "2.0":
            print("警告: xlrd>=2.0不再支持xls格式，请安装1.2.0版本: pip install xlrd==1.2.0")

    def _memory_monitor(self):
        """内存监控守护线程"""
        while not self.shutdown_flag:
            try:
                mem = psutil.virtual_memory()
                if mem.percent >= 85:  # 系统内存不足15%
                    self.event_queue.put(f"内存不足! 使用率: {mem.percent}%, 开始终止运行时间最长的进程...")
                    
                    # 找出运行时间最长的进程
                    longest_proc = None
                    max_runtime = 0
                    
                    for p in multiprocessing.active_children():
                        proc = psutil.Process(p.pid)
                        runtime = time.time() - proc.create_time()
                        if runtime > max_runtime:
                            max_runtime = runtime
                            longest_proc = p
                    
                    # 终止运行时间最长的进程
                    if longest_proc:
                        self.event_queue.put(
                            f"终止进程 {longest_proc.name} (PID: {longest_proc.pid}), "
                            f"已运行 {max_runtime:.1f} 秒"
                        )
                        try:
                            longest_proc.terminate()
                        except Exception as e:
                            self.event_queue.put(f"终止进程失败: {str(e)}")
                    
                    # 强制清理资源
                    self._kill_libreoffice_processes()
                    
                    # 等待内存恢复
                    time.sleep(10)
            
            except Exception as e:
                self.event_queue.put(f"内存监控错误: {str(e)}")
            
            # 每10秒检查一次内存
            time.sleep(10)

    def setup_signal_handlers(self):
        signal.signal(signal.SIGTERM, self.handle_shutdown)
        signal.signal(signal.SIGINT, self.handle_shutdown)

    def handle_shutdown(self, signum, frame):
        if not self.shutdown_flag:
            self.shutdown_flag = True
            self.shutdown_event.set()
            print("\n接收到终止信号，正在停止工作进程...")
            print("请等待当前任务完成或再次按Ctrl+C强制退出")
        else:
            self.force_shutdown = True
            print("\n强制终止所有进程...")
            self._force_terminate()
            sys.exit(1)
    
    def _force_terminate(self):
        """强制终止所有子进程和工作进程"""
        for child in multiprocessing.active_children():
            if child.is_alive():
                print(f"强制终止工作进程 {child.name} (PID: {child.pid})")
                try:
                    child.terminate()
                except Exception as e:
                    print(f"终止进程失败: {str(e)}")
        
        # 终止LibreOffice进程
        self._kill_libreoffice_processes()
        
        time.sleep(1)
    
    def _kill_libreoffice_processes(self):
        """递归终止所有LibreOffice相关进程"""
        for proc in psutil.process_iter(['name', 'pid', 'cmdline']):
            try:
                if proc.info['name'] == 'soffice.bin' or \
                   (proc.info['cmdline'] and 'libreoffice' in ' '.join(proc.info['cmdline'])):
                    print(f"强制终止LibreOffice进程 (PID: {proc.pid})")
                    self._kill_process_tree(proc.pid)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
                
    def _kill_process_tree(self, pid):
        """递归终止进程及其所有子进程"""
        try:
            parent = psutil.Process(pid)
            children = parent.children(recursive=True)
            for child in children:
                try:
                    child.terminate()
                except psutil.NoSuchProcess:
                    pass
            parent.terminate()
            
            # 等待进程结束
            gone, alive = psutil.wait_procs([parent] + children, timeout=5)
            for p in alive:
                try:
                    p.kill()
                except:
                    pass
        except psutil.NoSuchProcess:
            pass
        except Exception as e:
            print(f"终止进程树失败 (PID: {pid}): {str(e)}")
                
    def init_elasticsearch(self):
        try:
            es = Elasticsearch(
                hosts=[ES_HOST],
                api_key=ES_API_KEY,
                ca_certs=ES_CA_CERTS,
                request_timeout=60,
                retry_on_timeout=True,
                max_retries=3
            )
            if not es.ping():
                raise RuntimeError("无法连接到Elasticsearch")
            return es
        except Exception as e:
            print(f"警告: Elasticsearch初始化失败: {str(e)}")
            return None

    @contextlib.contextmanager
    def es_connection(self):
        if self.es_client is None:
            try:
                self.es_client = self.init_elasticsearch()
            except Exception as e:
                self.event_queue.put(f"ES重新连接失败: {str(e)}")
    
        retries = 0
        last_error = None
        while retries < MAX_ES_RETRIES and not self.shutdown_flag:
            try:
                if self.es_client is not None and self.es_client.ping():
                    try:
                        yield self.es_client
                        return
                    finally:
                        pass  # 确保生成器正确关闭
                else:
                    raise RuntimeError("ES连接不可用")
            except Exception as e:
                last_error = e
                self.event_queue.put(f"ES操作失败(尝试 {retries+1}/{MAX_ES_RETRIES}): {str(e)}")
                retries += 1
                if retries < MAX_ES_RETRIES:
                    time.sleep(ES_RETRY_DELAY)
                # 每次重试前尝试重新连接
                try:
                    self.es_client = self.init_elasticsearch()
                except:
                    pass
        
        if self.shutdown_flag:
            yield None
        else:
            raise RuntimeError(f"ES操作超过最大重试次数: {str(last_error)}")

    def check_es_health(self):
        """检查ES连接健康状况"""
        try:
            with self.es_connection() as es:
                if es is None:
                    return False
                health = es.cluster.health()
                if health['status'] in ['red', 'yellow']:
                    self.event_queue.put(f"ES集群状态不佳: {health['status']}")
                    return False
                return True
        except Exception as e:
            self.event_queue.put(f"ES健康检查失败: {str(e)}")
            return False

    def run(self):
        """主运行方法"""
        try:
            print("\n=== 文件处理阶段开始 ===")
            self.start_time = time.time()
            print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
            
            # 检查ES健康状况
            if not self.check_es_health():
                print("警告: Elasticsearch连接存在问题，处理可能会受到影响")
            
            # 重置卡住的任务
            self.event_queue.put("重置卡住的任务状态...")
            self._reset_stuck_tasks()
            time.sleep(1)
            
            # 获取初始待处理文件数
            total_pending = self._get_pending_count()
            with self.total_tasks.get_lock():
                self.total_tasks.value = total_pending
                
            if total_pending == 0:
                print("没有待处理文件")
                return
            
            # 启动任务分发线程
            self.dispatcher_thread = threading.Thread(
                target=self._task_dispatcher,
                daemon=True
            )
            self.dispatcher_thread.start()
            self.event_queue.put("任务分发线程已启动")
            
            # 启动结果处理线程
            self.result_handler_thread = threading.Thread(
                target=self._result_handler,
                daemon=True
            )
            self.result_handler_thread.start()
            self.event_queue.put("结果处理线程已启动")
            
            # 启动进度监控线程
            self.progress_thread = threading.Thread(
                target=self._progress_monitor,
                daemon=True
            )
            self.progress_thread.start()
            
            # 创建工作进程池
            workers = self._start_worker_pool()
            
            # 主监控循环
            while not self.shutdown_flag:
                # 检查工作进程状态
                self._check_worker_status(workers)
                
                # 检查处理完成情况
                if self._is_processing_complete():
                    self.event_queue.put("所有文件处理完成!")
                    break
                    
                time.sleep(1)
                
            # 等待工作进程结束
            self._stop_workers(workers)
            
            # 等待结果处理完成
            self.shutdown_event.set()
            if self.result_handler_thread and self.result_handler_thread.is_alive():
                self.result_handler_thread.join(timeout=10)
                
            # 生成最终报告
            if not self.shutdown_flag:
                self._print_final_report()
                
        except Exception as e:
            print(f"[主进程] 错误: {str(e)}")
            traceback.print_exc()
        finally:
            self._cleanup_resources()
            print("\n文件处理结束")

    def _progress_monitor(self):
        """简约进度监控线程"""
        last_display_time = 0
        recent_events = collections.deque(maxlen=3)  # 保存最近3个事件
        
        try:
            while not self.shutdown_flag:
                current_time = time.time()
                
                # 每1秒更新一次显示
                if current_time - last_display_time < 1.0:
                    time.sleep(0.1)
                    continue
                    
                last_display_time = current_time
                
                # 获取最新事件
                try:
                    while not self.event_queue.empty():
                        event = self.event_queue.get_nowait()
                        recent_events.append(event)
                except:
                    pass
                
                # 获取当前处理状态
                total = self.total_tasks.value
                completed = self.completed_tasks.value
                success = self.success_tasks.value
                failed = self.failed_tasks.value
                pending = max(0, total - completed)  # 避免负值
                
                # 计算处理速度
                if last_display_time != self.start_time:
                    time_elapsed = current_time - self.start_time
                    speed_per_min = completed / max(1, time_elapsed / 60)  # 每分钟处理速度
                else:
                    speed_per_min = 0
                
                # 计算运行时间
                runtime = current_time - self.start_time
                hours, rem = divmod(runtime, 3600)
                minutes, seconds = divmod(rem, 60)
                runtime_str = f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}"
                
                # 获取队列大小
                try:
                    task_qsize = self.task_queue.qsize()
                    result_qsize = self.result_queue.qsize()
                except:
                    task_qsize = 0
                    result_qsize = 0
                
                # 构建显示内容 - 简约设计
                display_lines = [
                    f"开始时间: {datetime.fromtimestamp(self.start_time).strftime('%Y-%m-%d %H:%M:%S')}",
                    f"已运行: {runtime_str} | 速度: {speed_per_min:.1f} 文件/分钟",
                    f"进度: 已完成 {completed}/{total} | 成功: {success} | 失败: {failed}",
                    f"队列: [待处理: {task_qsize} | 结果: {result_qsize}]",
                    ""
                ]
                
                # 添加最近事件
                if recent_events:
                    display_lines.append("最近事件:")
                    for event in recent_events:
                        display_lines.append(f"  - {event}")
                
                # 清屏并显示新内容 (兼容Linux/macOS和Windows)
                self._clear_screen()
                print("\n=== 文件处理进度 ===")
                print("\n".join(display_lines))
                
                # 检查处理是否完成
                if self._is_processing_complete():
                    break
                    
        except Exception as e:
            print(f"进度监控错误: {str(e)}")
        finally:
            # 确保最后一次显示最终状态
            self._clear_screen()
            print("\n=== 文件处理进度 ===")
            print("所有任务已完成，正在生成报告...")
            
    def _clear_screen(self):
        """清屏函数，兼容不同操作系统"""
        try:
            if os.name == 'posix':  # Linux and macOS
                os.system('clear')
            elif os.name == 'nt':    # Windows
                os.system('cls')
            else:
                print('\n' * 100)  # 其他系统使用空行模拟清屏
        except:
            print('\n' * 100)  # 出错时使用空行模拟清屏

    def _reset_stuck_tasks(self):
        try:
            with self.es_connection() as es:
                if es is None:  # 在关闭状态下不执行
                    return

                # 计算超时时间点
                reset_time = datetime.utcnow() - timedelta(seconds=LOCK_TIMEOUT)

                # 查询条件：匹配 processing（带 lock_time 检查）或 failed（无条件）
                query = {
                    "bool": {
                        "should": [
                            {
                                "bool": {
                                    "must": [
                                        {"term": {"status": "processing"}},
                                        {"range": {"lock_time": {"lte": reset_time.isoformat()}}}
                                    ]
                                }
                            },
                            {
                                #"term": {"status": "failed"}
                            }
                        ],
                        "minimum_should_match": 1  # 至少匹配一个条件
                    }
                }

                # 更新脚本：重置状态并清除 lock_time
                script = {
                    "source": """
                        ctx._source.status = 'pending';
                        ctx._source.lock_time = null;
                        if (ctx._source.containsKey('error_message')) {
                            ctx._source.remove('error_message');
                        }
                    """,
                    "lang": "painless"
                }

                # 执行更新
                res = es.update_by_query(
                    index=self.file_hash_index_name,
                    body={
                        "query": query,
                        "script": script
                    },
                    conflicts="proceed",
                    timeout="60s"
                )

                if res.get("updated", 0) > 0:
                    self.event_queue.put(f"已重置 {res['updated']} 个卡住的任务（processing 和 failed 状态）")
        except Exception as e:
            self.event_queue.put(f"重置任务状态失败: {str(e)}")
            
    def _get_pending_count(self) -> int:
        try:
            with self.es_connection() as es:
                if es is None:  # 在关闭状态下不执行
                    return 0
                    
                res = es.count(
                    index=self.file_hash_index_name,
                    body={
                        "query": {
                            "bool": {
                                "must": [
                                    {"term": {"status": "pending"}},
                                    {"terms": {"file_ext": list(SUPPORTED_EXTS)}},
                                    {"prefix": {"primary_file": "/服务器/2.教学部门/"}}
                                ]
                            }
                        }
                    }
                )
                return res["count"]
        except Exception as e:
            self.event_queue.put(f"获取待处理文件数失败: {str(e)}")
            return 0

    def _start_worker_pool(self) -> list:
        physical_cores = multiprocessing.cpu_count()  # 使用逻辑核心数而不是物理核心
        num_workers = max(1, int(physical_cores * CPU_USEAGE))
        
        workers = []
        for i in range(num_workers):
            worker_id = f"Worker-{self._next_worker_id}"
            self._next_worker_id += 1
            
            p = multiprocessing.Process(
                target=self.process_worker,
                args=(self.shutdown_event, worker_id),
                daemon=False,
                name=worker_id
            )
            p.start()
            workers.append(p)
            self.event_queue.put(f"启动工作进程 #{worker_id} (PID: {p.pid})")
        
        return workers

    def _task_dispatcher(self):
        """任务分发线程 - 主进程运行"""
        try:
            while not self.shutdown_flag:
                # 检查任务队列状态
                if self.task_queue.qsize() < TASK_QUEUE_SIZE / 2:
                    # 填充任务队列
                    self._fill_task_queue()
                    
                time.sleep(0.5)
                
        except Exception as e:
            self.event_queue.put(f"任务分发线程错误: {str(e)}")
        finally:
            # 发送毒丸信号通知工作进程退出
            self.event_queue.put("任务分发线程已退出")
            for _ in range(10):  # 确保所有工作进程都收到毒丸
                try:
                    self.task_queue.put(None, timeout=1)
                except queue.Full:
                    pass
    
    def _fill_task_queue(self):
        """从ES获取一批任务并放入任务队列"""
        try:
            with self.es_connection() as es:
                if es is None or self.shutdown_flag:
                    return
                
                # 搜索待处理任务
                search_body = {
                    "query": {
                        "bool": {
                            "must": [
                                {"term": {"status": "pending"}},
                                {"terms": {"file_ext": list(SUPPORTED_EXTS)}},
                                {"prefix": {"primary_file": "/服务器/2.教学部门/"}}
                            ]
                        }
                    },
                    "size": TASK_BATCH_SIZE,
                    "_source": ["primary_file", "file_ext", "modified_date", "retry_count"],
                    "sort": [{"modified_date": "asc"}]
                }
                
                try:
                    res = es.search(
                        index=self.file_hash_index_name,
                        body=search_body,
                        request_timeout=30
                    )
                    
                    hits = res.get("hits", {}).get("hits", [])
                    if not hits:
                        time.sleep(1)  # 没有任务时稍作休息
                        return
                    
                    # 转换为列表避免生成器问题
                    hits_list = list(hits)
                    
                    locked_docs = []
                    for hit in hits_list:
                        doc_id = hit["_id"]
                        source = hit["_source"]
                        
                        try:
                            update_result = es.update(
                                index=self.file_hash_index_name,
                                id=doc_id,
                                body={
                                    "doc": {
                                        "status": "processing",
                                        "lock_time": datetime.utcnow().isoformat()
                                    }
                                },
                                retry_on_conflict=0
                            )
                            
                            if update_result["result"] == "updated":
                                locked_docs.append((doc_id, source))
                        except:
                            continue  # 如果失败则跳过
                    
                    if not locked_docs:
                        return
                        
                    # 将锁定的任务加入任务队列
                    added_count = 0
                    for doc_id, source in locked_docs:
                        try:
                            self.task_queue.put((doc_id, source), timeout=1)
                            added_count += 1
                            # 更新任务计数器
                            with self.task_counter.get_lock():
                                self.task_counter.value += 1
                        except queue.Full:
                            # 如果队列满，将剩余任务释放
                            self._release_tasks(es, locked_docs[added_count:])
                            break
                    
                    if added_count > 0:
                        self.event_queue.put(f"已添加 {added_count} 个新任务到队列")
                    
                except Exception as search_error:
                    self.event_queue.put(f"ES搜索失败: {str(search_error)}")
                    time.sleep(5)  # 等待后重试
                    
        except Exception as e:
            if not self.shutdown_flag:  # 非关闭状态下才报告错误
                self.event_queue.put(f"填充任务队列失败: {str(e)}")
                time.sleep(5)  # 添加延迟防止快速重试
            
    def _release_tasks(self, es, tasks):
        """释放未加入队列的任务"""
        try:
            for doc_id, _ in tasks:
                try:
                    es.update(
                        index=self.file_hash_index_name,
                        id=doc_id,
                        body={
                            "doc": {
                                "status": "pending",
                                "lock_time": None
                            }
                        }
                    )
                except:
                    continue
        except Exception as e:
            self.event_queue.put(f"释放任务失败: {str(e)}")
            
    def _result_handler(self):
        """结果处理线程 - 主进程运行"""
        try:
            while not self.shutdown_flag or self.task_counter.value > 0:
                try:
                    # 等待结果，设置超时避免永久阻塞
                    result = self.result_queue.get(timeout=5)
                    
                    # 解析结果
                    worker_id, doc_id, status, content, error_msg = result
                    
                    # 如果是毒丸信号或退出信号，跳过更新
                    if doc_id is None:
                        continue
                    
                    try:
                        # 更新文档状态
                        with self.es_connection() as es:
                            if es:
                                if status == "completed":
                                    update_body = {
                                        "doc": {
                                        "status": "completed",
                                        "lock_time": None,
                                        "content": content
                                        }
                                    }
                                    self.event_queue.put(f"成功处理文档: {doc_id}")
                                else:
                                    update_body = {
                                        "doc": {
                                            "status": "failed",
                                            "lock_time": None,
                                            "error_message": error_msg
                                        }
                                    }
                                    self.event_queue.put(f"处理失败: {doc_id} - {error_msg[:60]}")
                                    
                                es.update(
                                    index=self.file_hash_index_name,
                                    id=doc_id,
                                    body=update_body,
                                    timeout="30s"
                                )
                        
                        # 更新状态计数器
                        with self.completed_tasks.get_lock():
                            self.completed_tasks.value += 1
                            
                        if status == "completed":
                            with self.success_tasks.get_lock():
                                self.success_tasks.value += 1
                        else:
                            with self.failed_tasks.get_lock():
                                self.failed_tasks.value += 1
                        
                    except Exception as update_err:
                        self.event_queue.put(f"更新文档状态失败: {str(update_err)}")
                        
                    # 更新任务计数器
                    if self.task_counter.value > 0:
                        with self.task_counter.get_lock():
                            self.task_counter.value -= 1
                            
                except queue.Empty:
                    if self.shutdown_flag and self.task_counter.value == 0:
                        break
                    continue
                    
        except Exception as e:
            self.event_queue.put(f"结果处理线程错误: {str(e)}")
            
    def _is_processing_complete(self):
        """检查处理是否完成"""
        if self.task_counter.value > 0:
            return False
            
        pending_count = self._get_pending_count()
        if pending_count > 0:
            return False
            
        # 检查是否有未处理的任务
        if not self.task_queue.empty() or not self.result_queue.empty():
            return False
            
        return True

    def _check_worker_status(self, workers: list):
        """检查并维护工作进程池"""
        for p in workers[:]:
            if not p.is_alive():
                workers.remove(p)
                self.event_queue.put(f"工作进程 {p.name} 已退出 (退出码: {p.exitcode})")
                
                # 正常退出时重启进程
                if p.exitcode == 0 and not self.shutdown_flag:
                    worker_id = f"Worker-{self._next_worker_id}"
                    self._next_worker_id += 1
                    
                    new_p = multiprocessing.Process(
                        target=self.process_worker,
                        args=(self.shutdown_event, worker_id),
                        daemon=False,
                        name=worker_id
                    )
                    new_p.start()
                    workers.append(new_p)
                    self.event_queue.put(f"重启工作进程 #{worker_id} (PID: {new_p.pid})")

    def _stop_workers(self, workers: list):
        if self.force_shutdown:
            return

        self.event_queue.put("正在停止工作进程...")
        self.shutdown_event.set()
        
        # 设置进程超时
        timeout = 10
        start = time.time()
        
        for p in workers:
            remaining = max(0, timeout - (time.time() - start))
            p.join(timeout=remaining)
            
            if p.is_alive():
                self.event_queue.put(f"强制终止工作进程 {p.name} (PID: {p.pid})")
                try:
                    p.terminate()
                except Exception as e:
                    self.event_queue.put(f"终止进程失败: {str(e)}")
        
        self.event_queue.put("所有工作进程已停止")

    def _print_final_report(self):
        """生成简约的最终报告"""
        try:
            elapsed = time.time() - self.start_time
            elapsed_time = timedelta(seconds=elapsed)
            minutes, seconds = divmod(elapsed, 60)
            hours, minutes = divmod(minutes, 60)
            formatted_time = f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}"
            
            total = self.total_tasks.value
            completed = self.completed_tasks.value
            success = self.success_tasks.value
            fail = self.failed_tasks.value
            
            # 检查除以零的情况
            success_rate = (success / completed * 100) if completed > 0 else 0.0
            
            report = [
                "\n" + "=" * 50,
                "文件处理完成报告",
                "=" * 50,
                f"总运行时间: {formatted_time}",
                f"初始文件总数: {total}",
                f"已处理文件数: {completed}",
                f"  成功: {success}",
                f"  失败: {fail}",
                f"处理成功率: {success_rate:.1f}%",
                "=" * 50
            ]
            print("\n".join(report))
            
        except Exception as e:
            print(f"生成最终报告失败: {str(e)}")

    def _cleanup_resources(self):
        """清理资源"""
        self.shutdown_event.set()
        
        # 等待线程结束
        if self.progress_thread and self.progress_thread.is_alive():
            self.progress_thread.join(timeout=1)
        if self.dispatcher_thread and self.dispatcher_thread.is_alive():
            self.dispatcher_thread.join(timeout=1)
        if self.result_handler_thread and self.result_handler_thread.is_alive():
            self.result_handler_thread.join(timeout=1)
        
        # 强制终止任何残留的LibreOffice进程
        self._kill_libreoffice_processes()

    def process_worker(self, shutdown_event: Event, worker_id: str):
        """工作进程主函数"""
        # 设置信号处理
        signal.signal(signal.SIGTERM, signal.SIG_DFL)
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        
        # 初始化状态
        ocr = None
        processed_count = 0
        
        try:
            while not shutdown_event.is_set() and processed_count < TASKS_PER_WORKER:
                try:
                    # 获取任务，带超时
                    try:
                        task = self.task_queue.get(timeout=5)
                    except queue.Empty:
                        # 如果没有任务且收到关闭信号，退出
                        if shutdown_event.is_set():
                            break
                        continue
                        
                    # 处理毒丸信号
                    if task is None:
                        self.result_queue.put((worker_id, None, None, None, "毒丸信号"))
                        break
                    
                    doc_id, file_record = task
                    file_path = Path(file_record["primary_file"])
                    ext = file_record["file_ext"].lower()
                    
                    self.event_queue.put(f"[{worker_id}] 处理: {file_path.name}")
                    
                    # 按需初始化OCR
                    if ext in ('.pdf', '.ppt', '.pptx') and ocr is None:
                        try:
                            ocr = PaddleOCR(use_angle_cls=True, lang="ch", use_gpu=False, show_log=False)
                            self.event_queue.put(f"[{worker_id}] OCR初始化成功")
                        except Exception as e:
                            error_msg = f"OCR初始化失败: {str(e)}"
                            self.event_queue.put(f"[{worker_id}] {error_msg}")
                            self.result_queue.put((worker_id, doc_id, "failed", None, error_msg))
                            continue
                    
                    try:
                        if ext in ('.xlsx', '.xls'):
                            content = self._process_excel(file_path)
                        elif ext == '.docx':
                            content = self._process_docx(file_path)
                        elif ext == '.doc':
                            content = self._process_doc(file_path)
                        elif ext == '.pdf':
                            content = self._process_pdf(ocr, file_path)
                        elif ext in ('.ppt', '.pptx'):
                            content = self._process_ppt(file_path, ocr)
                        elif ext in ('.html', '.htm', '.txt'):
                            content = self._process_text(file_path)
                        else:
                            content = None
                            error_msg = f"不支持的文件类型: {ext}"
                            
                        # 发送处理结果
                        if content:
                            self.result_queue.put((worker_id, doc_id, "completed", content, None))
                        else:
                            self.result_queue.put((worker_id, doc_id, "failed", None, error_msg or "内容提取失败"))
                        
                        # 增加处理计数
                        processed_count += 1
                        
                        # 达到处理上限后退出
                        if processed_count >= TASKS_PER_WORKER:
                            self.event_queue.put(f"[{worker_id}] 已完成 {TASKS_PER_WORKER} 个任务，准备重启")
                            break
                        
                    except Exception as e:
                        error_msg = str(e)[:500]
                        self.event_queue.put(f"[{worker_id}] 处理失败: {str(e)[:100]}")
                        self.result_queue.put((worker_id, doc_id, "failed", None, error_msg))
                        
                except Exception as outer_e:
                    error_msg = f"工作进程内部错误: {str(outer_e)}"
                    self.event_queue.put(f"[{worker_id}] 内部错误: {str(outer_e)[:100]}")
                    if 'doc_id' in locals():
                        self.result_queue.put((worker_id, doc_id, "failed", None, error_msg))
                    
        except Exception as e:
            error_msg = f"工作进程异常: {str(e)}"
            self.event_queue.put(f"[{worker_id}] 异常: {str(e)[:100]}")
        finally:
            # 清理资源
            if ocr:
                try:
                    if hasattr(ocr, 'ocr_engine'):
                        ocr.ocr_engine = None
                    if hasattr(ocr, 'text_system'):
                        ocr.text_system = None
                    del ocr
                except:
                    pass
            
            # 发送结束信号
            try:
                self.result_queue.put((worker_id, None, "exit", None, "工作进程退出"), timeout=1)
            except:
                pass
            
            sys.exit(0)

    def _process_excel(self, file_path: Path):
        wb = None
        full_content = []
        ext = file_path.suffix.lower()
        
        try:
            if ext == '.xlsx':
                wb = openpyxl.load_workbook(file_path, data_only=True)
                for sheet_name in wb.sheetnames:
                    sheet = wb[sheet_name]
                    sheet_content = []
                    for row in sheet.iter_rows(values_only=True):
                        row_content = []
                        for cell in row:
                            if cell is None:
                                row_content.append("")
                            else:
                                if isinstance(cell, datetime):
                                    row_content.append(cell.strftime('%Y-%m-%d %H:%M:%S'))
                                else:
                                    row_content.append(str(cell).strip())
                        sheet_content.append("\t".join(row_content))
                    full_content.append(f"=== 工作表: {sheet_name} ===\n" + "\n".join(sheet_content))
                    
            elif ext == '.xls':
                wb = xlrd.open_workbook(file_path)
                for sheet_name in wb.sheet_names():
                    sheet = wb.sheet_by_name(sheet_name)
                    sheet_content = []
                    for row_idx in range(sheet.nrows):
                        row_content = []
                        for col_idx in range(sheet.ncols):
                            cell = sheet.cell(row_idx, col_idx)
                            if cell.ctype == xlrd.XL_CELL_DATE:
                                try:
                                    cell_value = xldate_as_datetime(cell.value, wb.datemode).strftime('%Y-%m-%d %H:%M:%S')
                                except:
                                    cell_value = str(cell.value)
                            elif cell.ctype == xlrd.XL_CELL_NUMBER:
                                cell_value = int(cell.value) if cell.value.is_integer() else cell.value
                            else:
                                cell_value = cell.value
                            row_content.append(str(cell_value).strip())
                        sheet_content.append("\t".join(row_content))
                    full_content.append(f"=== 工作表: {sheet_name} ===\n" + "\n".join(sheet_content))
            
            return "\n\n".join(full_content)
        except Exception as e:
            raise RuntimeError(f"Excel处理失败: {str(e)}")
        finally:
            if wb:
                try:
                    if ext == '.xlsx':
                        wb.close()
                    elif ext == '.xls':
                        wb.release_resources()
                except:
                    pass

    def _process_docx(self, file_path: Path):
        try:
            doc = Document(file_path)
            full_text = []
            
            for para in doc.paragraphs:
                if para.text.strip():
                    full_text.append(para.text)
                
            for table in doc.tables:
                for row in table.rows:
                    row_text = []
                    for cell in row.cells:
                        if cell.text.strip():
                            row_text.append(cell.text)
                    if row_text:
                        full_text.append("\t".join(row_text))
            
            return "\n".join(full_text)
        except Exception as e:
            raise RuntimeError(f"DOCX处理失败: {str(e)}")

    def _process_doc(self, file_path: Path):
        """处理DOC文件，使用独立的用户目录和资源清理"""
        temp_dir = None
        output_path = None
        user_dir = None
        subproc = None
        
        try:
            temp_dir = Path(tempfile.mkdtemp(prefix="doc_"))
            user_dir = Path(tempfile.mkdtemp(prefix="lo_user_"))
            output_path = temp_dir / f"{file_path.stem}.txt"
            
            # 使用独立的用户目录和HOME环境变量
            user_dir_str = str(user_dir).replace('\\', '/')
            cmd = [
                'libreoffice',
                '--headless',
                '--convert-to', 'txt:Text (encoded):UTF8',
                '--outdir', str(temp_dir),
                '--nologo',
                '--norestore',
                '--nodefault',
                '--nolockcheck',
                f'-env:UserInstallation=file://{user_dir_str}',
                str(file_path)
            ]
            
            env = os.environ.copy()
            env["HOME"] = str(user_dir)  # 设置HOME环境变量避免权限问题
            
            subproc = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=env
            )
            
            try:
                _, _ = subproc.communicate(timeout=300)
                if subproc.returncode != 0:
                    raise RuntimeError(f"LibreOffice转换失败 (退出码: {subproc.returncode})")
                    
                if not output_path.exists():
                    raise RuntimeError("未生成输出文件")
                    
                content = output_path.read_text(encoding='utf-8', errors='replace')
                if not content.strip():
                    raise RuntimeError("转换结果为空")
                    
                return content
            except subprocess.TimeoutExpired:
                # 递归终止整个进程树
                self._kill_process_tree(subproc.pid)
                try:
                    # 尝试获取剩余输出
                    _, _ = subproc.communicate(timeout=1)
                except:
                    pass
                raise RuntimeError("DOC转换超时（超过5分钟）")
            except Exception as e:
                raise RuntimeError(f"DOC转换错误: {str(e)}")
                
        except Exception as e:
            raise RuntimeError(f"DOC处理失败: {str(e)}")
            
        finally:
            # 确保资源清理
            if output_path and output_path.exists():
                try:
                    output_path.unlink()
                except:
                    pass
                    
            # 确保临时目录和用户目录被清理
            if temp_dir:
                try:
                    shutil.rmtree(temp_dir, ignore_errors=True)
                except Exception as e:
                    print(f"清理临时目录失败: {str(e)}")
            
            if user_dir:
                try:
                    shutil.rmtree(user_dir, ignore_errors=True)
                except Exception as e:
                    print(f"清理用户目录失败: {str(e)}")
                
    def _process_ppt(self, file_path: Path, ocr=None):
        """处理PPT文件，使用独立的用户目录和资源清理"""
        temp_dir = None
        pdf_path = None
        user_dir = None
        subproc = None
        
        try:
            temp_dir = Path(tempfile.mkdtemp(prefix="ppt_"))
            user_dir = Path(tempfile.mkdtemp(prefix="lo_user_"))
            pdf_path = temp_dir / f"{file_path.stem}.pdf"
            
            # 使用独立的用户目录和HOME环境变量
            user_dir_str = str(user_dir).replace('\\', '/')
            cmd = [
                'libreoffice',
                '--headless',
                '--invisible',
                '--nodefault',
                '--nologo',
                '--norestore',
                '--nolockcheck',
                f'-env:UserInstallation=file://{user_dir_str}',
                '--convert-to', 'pdf',
                '--outdir', str(temp_dir),
                str(file_path)
            ]
            
            env = os.environ.copy()
            env["HOME"] = str(user_dir)  # 设置HOME环境变量避免权限问题
            
            subproc = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=env
            )
            
            try:
                _, _ = subproc.communicate(timeout=600)
                if subproc.returncode != 0:
                    raise RuntimeError(f"LibreOffice转换失败 (退出码: {subproc.returncode})")
                    
                if not pdf_path.exists():
                    raise RuntimeError("未生成PDF文件")
                    
                # 处理生成的PDF
                return self._process_pdf(ocr, pdf_path)
                    
            except subprocess.TimeoutExpired:
                # 递归终止整个进程树
                self._kill_process_tree(subproc.pid)
                try:
                    # 尝试获取剩余输出
                    _, _ = subproc.communicate(timeout=1)
                except:
                    pass
                raise RuntimeError("PPT转换超时（超过10分钟）")
            except Exception as e:
                raise RuntimeError(f"PPT转换错误: {str(e)}")
                
        except Exception as e:
            raise RuntimeError(f"PPT处理失败: {str(e)}")
            
        finally:
            # 确保资源清理
            if pdf_path and pdf_path.exists():
                try:
                    pdf_path.unlink()
                except:
                    pass
                    
            # 确保临时目录和用户目录被清理
            if temp_dir:
                try:
                    shutil.rmtree(temp_dir, ignore_errors=True)
                except Exception as e:
                    print(f"清理临时目录失败: {str(e)}")
            
            if user_dir:
                try:
                    shutil.rmtree(user_dir, ignore_errors=True)
                except Exception as e:
                    print(f"清理用户目录失败: {str(e)}")
                
    def _process_pdf(self, ocr: PaddleOCR, file_path: Path):
        # 设置40分钟超时
        start_time = time.time()
        timeout = 2400  # 40分钟

        try:
            with fitz.open(str(file_path)) as doc:
                full_text = []
                for page_num in range(len(doc)):
                    # 检查超时
                    if time.time() - start_time > timeout:
                        raise TimeoutError("PDF处理超时（超过40分钟）")

                    try:
                        page = doc.load_page(page_num)
                        
                        # 尝试直接提取文本
                        try:
                            text = page.get_text()
                            if text and text.strip():
                                full_text.append(text)
                                continue
                        except:
                            pass
                        
                        try:
                            # 对于扫描的PDF，使用OCR
                            pix = page.get_pixmap(dpi=150)
                            img = Image.frombytes("RGB", (pix.width, pix.height), pix.samples)
                            
                            img_np = np.array(img).astype(np.uint8)
                            if img_np.ndim == 2:
                                img_np = cv2.cvtColor(img_np, cv2.COLOR_GRAY2RGB)
                            elif img_np.shape[2] == 4:
                                img_np = cv2.cvtColor(img_np, cv2.COLOR_RGBA2RGB)
                                
                            result = ocr.ocr(img_np, cls=False)
                            
                            ocr_text = []
                            if result and result[0]:
                                for line in result[0]:
                                    if line and len(line) >= 2 and line[1]:
                                        text_content = str(line[1][0])
                                        if text_content.strip():
                                            ocr_text.append(text_content)
                            
                            full_text.append("\n".join(ocr_text) if ocr_text else "")
                        except Exception as ocr_error:
                            # 如果单页处理失败，记录错误但继续处理下一页
                            full_text.append(f"[OCR处理失败] 页面 {page_num+1}: {str(ocr_error)}")
                    except Exception as page_error:
                        full_text.append(f"[页面处理失败] 页面 {page_num+1}: {str(page_error)}")
                
                return "\n".join(full_text)
        except TimeoutError as te:
            # 超时处理
            raise RuntimeError(str(te))
        except Exception as e:
            raise RuntimeError(f"PDF处理失败: {str(e)}")
                    
    def _process_text(self, file_path: Path):
        try:
            # 尝试常见编码
            encodings = ['utf-8', 'gbk', 'gb2312', 'gb18030', 'big5', 'latin-1']
            content = None
            
            for encoding in encodings:
                try:
                    with open(file_path, 'r', encoding=encoding) as f:
                        content = f.read()
                    if content.strip():  # 确保读取到内容
                        break
                except (UnicodeDecodeError, LookupError):
                    continue
            
            if content is None:
                # 所有编码尝试失败，使用二进制模式读取
                with open(file_path, 'rb') as f:
                    content = f.read().decode('latin-1', errors='replace')
            
            return content
        except Exception as e:
            raise RuntimeError(f"文本文件处理失败: {str(e)}")

def main():
    try:
        processor = ProcessPhase()
        processor.run()
    except KeyboardInterrupt:
        print("\n[提示] 用户中断操作")
    except Exception as e:
        print(f"\n[致命错误] {str(e)}")
        traceback.print_exc()
        sys.exit(1)

if __name__ == "__main__":
    multiprocessing.freeze_support()
    main()